package cc.example;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class 自定义线程和调度器 {
    public static void main(String[] args) {
        Flux.range(1, 10)
                .publishOn(Schedulers.boundedElastic()) // 在哪个线程池把这个流的数据和操作执行了
                .log()
                .map(integer -> integer + 10)
                .log();
//                .subscribe();

        // 调度器就是线程池
        Schedulers.immediate(); // 当前线程运行所有操作
        Schedulers.single(); // 使用固定的一个单线程
        Schedulers.boundedElastic(); // 有界,弹性调度

        Schedulers.fromExecutor(new ThreadPoolExecutor(4, 8, 50, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1000)));

        Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);

        final Flux<String> flux = Flux
                .range(1, 2)
                .map(i -> 10 + i)
                .log()
                .publishOn(s)
                .map(i -> "value " + i)
                .subscribeOn(Schedulers.newSingle("我啊你"))
                .log();

        new Thread(() -> flux.subscribe(System.out::println)).start();
    }


}
